package com.ahaha.myBy

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object combinerByKeyTest {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName(s"${combinerByKeyTest.getClass.getSimpleName}")
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
    //aggregateByKeyOps(sc)
    test1(sc)
    sc.stop()
  }

  def test1(sc:SparkContext): Unit ={
    val lines = sc.parallelize(List(
      "hello you",
      "hello me",
      "hello lan lan"
    ))

    val pairs = lines.flatMap(_.split("\\s+")).map((_, 1))

    /* createCombiner: V => C,
    mergeValue: (C, V) => C,
    mergeCombiners: (C, C) => C)*/

    def createCombiner(num: Int): Int ={
      num
    }

    def mergeValue(sum:Int,num:Int): Int ={
      sum + num
    }

    def mergeCombiners(sum1:Int,sum2:Int):Int={
      sum1+sum2
    }

    val res = pairs.combineByKey(createCombiner, mergeValue, mergeCombiners)

    println("combinerbykey后的结果")
    res.foreach(println)


  }

  def combineByKeyOps(sc:SparkContext): Unit ={
    case class Student(id:Int,name:String,province:String)

    var stuRDD = sc.parallelize(List(
      Student(1, "张三", "天津"),
      Student(4, "李四", "湖南"),
      Student(10, "王五", "天津"),
      Student(6, "赵六", "湖南"),
      Student(2, "田七", "山西"),
      Student(3, "赵八", "山东"),
      Student(7, "孙九", "山西"),
      Student(5, "郑十", "湖南"),
      Student(10087, "熊大", "山西"),
      Student(8, "熊二", "山东"),
      Student(10086, "伍六七", "湖南"),
      Student(10089, "雷电俱散", "山东")
    ), 3)

    val p2stu:RDD[(String,Student)] = stuRDD.map(stu => (stu.province, stu))

   /* createCombiner: V => C,
    mergeValue: (C, V) => C,
    mergeCombiners: (C, C) => C)*/
    def createCombiner(stu:Student):Array[Student]={
      println("根据key创建分区")
      Array(stu)
    }

    def mergeValue(stus:Array[Student],stu:Student):Array[Student] = {
      println("同一个分区下的相同key的聚合")
      stus.+:(stu)
    }

    def mergeCombiners(stus1:Array[Student],stus2:Array[Student]): Array[Student] ={
      println("")
      stus1 ++ stus2
    }

    val res = p2stu.combineByKey(createCombiner, mergeValue, mergeCombiners)

    res.foreach{ case (province, students) => {
      println(s"${province}对应的学生有：${students.toList}")
    }}
  }

}
